#include "TcpConnection.h"
#include <unistd.h>


using namespace tmms::network;

TcpConnection::TcpConnection(EventLoop *loop,
                int socketfd,
                const InetAddress &localAddr,
                const InetAddress& peerAddr)
:Connection(loop,socketfd,localAddr,peerAddr)
{

}

TcpConnection::~TcpConnection()
{
    OnClose();
}

void TcpConnection::OnClose()
{
    loop_->AssertInLoopThread();
    if(!closed_)
    {
        if(close_cb_)
        {
            close_cb_(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()));
        }
        closed_ = true;
        Event::Close();
    }
}

void TcpConnection::ForceClose() //他说什么可能是其他线程调用的说要放到事件循坏中？？？
{
    loop_->RunInLoop([this](){
        OnClose();
    });
}

void TcpConnection::setCloseCallback(const CloseConnectionCallback &cb)
{
    close_cb_= cb;
}

void TcpConnection::setCloseCallback( CloseConnectionCallback &&cb)
{
    close_cb_=std::move(cb) ;
}



void TcpConnection::setRecvMsgCallback(const MessageCallback &cb)
{
    message_cb_ = cb;
}


void TcpConnection::setRecvMsgCallback(MessageCallback &&cb)
{
    message_cb_ = std::move(cb);
}

void TcpConnection::OnRead()
{
    if(closed_)
    {
        NEWORK_TRACE<<"host :"<<peer_addr_.ToIpPort()<<"had closed.";
        return ;
    }
    while (true)
    {
        int err =0;
        auto ret  = message_buffer_.readFd(fd_,&err);
        if(ret >0){
            if(message_cb_)
            {
                message_cb_(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()),message_buffer_);
            }
        }else if(ret  == 0)
        {
            OnClose();
            break;
        }else{
            if(err != EINTR && err!= EAGAIN && err != EWOULDBLOCK)
            {
                NEWORK_TRACE<<"read err:"<<err;
                OnClose();
            }
            break;
        }
    }  
}


void TcpConnection::OnError(const std::string &msg) 
{
    NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"error msg:"<<msg;
    OnClose();
} 


void TcpConnection::OnWrite() 
{
    if(closed_)
    {
       NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"had closed.";
       return ;
    }
    if(io_vec_list_.empty())
    {
        EnableWriting(false);
        if(write_complete_cb_)
        {
            write_complete_cb_(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()));
        }
        return ;
    }
    if(!io_vec_list_.empty())
    {
        while(true)
        {
            auto ret = ::writev(fd_,&io_vec_list_[0],io_vec_list_.size());
            if(ret >= 0)
            {
                while(ret>0)
                {
                    if(io_vec_list_.front().iov_len > ret){
                        //只发了一部分的数据
                        auto& front = io_vec_list_.front();
                        front.iov_base = static_cast<char*>(front.iov_base) + ret;
                        io_vec_list_.front().iov_len -= ret;
                        break;
                    }else{//发送全部了
                        ret -= io_vec_list_.front().iov_len;
                        io_vec_list_.erase(io_vec_list_.begin());
                    }
                }
                if(io_vec_list_.empty())
                {
                    EnableWriting(false);
                    if(write_complete_cb_)
                    {
                        write_complete_cb_(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()));
                    }
                    return ;
                }
            }else{ //结果小于0的情况
                if(errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)
                {
                    NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"write errno:"<<errno;
                    OnClose();
                    return ;
                }
                break;
            }
        }
    }
}


void TcpConnection::SetWriteCompleteCallback(const WriteCompleteCallback &cb)
{
    write_complete_cb_ = cb;
}


void TcpConnection::SetWriteCompleteCallback(const WriteCompleteCallback &&cb)
{
    write_complete_cb_ = std::move(cb);
}



void TcpConnection::Send(std::list<BufferNodePtr> &list)
{//因为发送的函数是在任何的线程之下都可能调用的，所以要把它放在我的事件循环之下
    loop_->RunInLoop([this,&list](){ //TODO: 声明周期由业务层去保证，所以要有个回调，等我发送完了，你就把数据清理掉
        SendInLoop(list);
    });
}

void TcpConnection::Send(const char *buf,size_t size)
{
    loop_->RunInLoop([this,buf,size](){ //TODO: 声明周期由业务层去保证，所以要有个回调，等我发送完了，你就把数据清理掉
        SendInLoop(buf,size);
    }); 
}


void  TcpConnection::SendInLoop(std::list<BufferNodePtr> &list)
{
    if(closed_)
    {
       NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"had closed.";
       return ;
    }
    for(auto &l:list)
    {
        struct iovec vec ;
        vec.iov_base =(void*)l->addr;
        vec.iov_len =l->size;

        io_vec_list_.push_back(vec);
    }
    if(!io_vec_list_.empty())
    {
        EnableWriting(true);
    }
}


void  TcpConnection::SendInLoop(const char *buf,size_t size)
{
    if(closed_)
    {
       NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"had closed.";
       return ;
    }
    size_t send_len =0;
    if(io_vec_list_.empty())
    {
        send_len = ::write(fd_,buf,size);
        if(send_len<0)
        {
            if(errno != EINTR && errno != EAGAIN && errno != EWOULDBLOCK)
            {
                NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"write errno:"<<errno;
                OnClose();
                return ;
            } 
            send_len = 0;
        }
        size -= send_len;
        if(size == 0)
        {//发送完了  我们直接回调
            if(write_complete_cb_)
            {
                write_complete_cb_(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()));
            }
            return ;
        }
    }
    if(size>0)
    {
        struct iovec vec;
        vec.iov_base = const_cast<char*>(buf + send_len);
        vec.iov_len = size;

        io_vec_list_.push_back(vec);
        EnableWriting(true);
    }
}

//超时事件
void TcpConnection::SetTimeoutCallback(int timeout,const TimeoutCallback &cb)
{
    auto cp = std::dynamic_pointer_cast<TcpConnection>(shared_from_this());
    loop_->RunAfter(timeout,[&cp,&cb](){
        cb(cp);
    });
}

void TcpConnection::SetTimeoutCallback(int timeout,TimeoutCallback &&cb)
{
    auto cp = std::dynamic_pointer_cast<TcpConnection>(shared_from_this());
    loop_->RunAfter(timeout,[&cp,cb](){
        cb(cp);
    });
}

void TcpConnection::OnTimeout()
{
    NEWORK_ERROR<<"host:"<<peer_addr_.ToIpPort()<<"timeout and close it.";
    OnClose();
}

void TcpConnection::EnableCheckIdleTimeout(int32_t max_time)
{
    auto tp = std::make_shared<TimeoutEntry>(std::dynamic_pointer_cast<TcpConnection>(shared_from_this()));
    max_idle_time_ = max_time;
    timeout_entry_ = tp;
    loop_->InserEntry(max_time,tp);
}

void TcpConnection::ExtendLife()
{
    auto tp = timeout_entry_.lock();
    if(tp)
    {
        loop_->InserEntry(max_idle_time_,tp);
    }
}



